package com.hs.kfk.transaction;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Description: 消息事务机制
 * @Author 胡尚
 * @Date: 2024/8/7 18:33
 */
public class TransactionProducer {

    private final static String BOOTSTRAP_SERVER = "192.168.75.61:9092,192.168.75.62:9092,192.168.75.63:9092";
    private final static String TOPIC_NAME = "disTopic";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 指定一个消息事务id, 可以随意指定
        properties.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "123");
        // 创建一个消息生产者对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        // 初始化事务 并 开启事务
        producer.initTransactions();
        producer.beginTransaction();

        try {
            for (int i = 0; i < 5; i++) {
                // 异步发送消息，当发送到id=3的消息时 抛异常
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, Integer.toString(i), "MyProducer" + i);
                producer.send(record);
                if (i == 3) {
                    throw new NullPointerException();
                }
            }
            // 提交事务
            producer.commitTransaction();
        } catch (Exception e) {
            // 回滚事务
            System.out.println("出异常啦！！！");
            producer.abortTransaction();
        } finally {
            producer.close();
        }
    }
}
